package com.imchenio.utils.parallel;

import com.imchenio.utils.parallel.ring.HashAlgorithm;
import com.imchenio.utils.parallel.ring.HashRing;
import com.imchenio.utils.parallel.runnable.ChannelRunnableFactory;

import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

public class ChannelGroup<T> {

    protected ConcurrentHashMap<Integer, ProcessChannel<T>> channels;
    protected HashAlgorithm ring;

    public ChannelGroup(int channelSize, int blockingQueueSize, ChannelRunnableFactory<T> factory) {
        assert channelSize > 0;
        channels = new ConcurrentHashMap<>();

        Set<String> randomStrings = new HashSet<>();
        for (int i = 0; i < channelSize; ++i) {
           String randomString = UUID.randomUUID().toString();
           randomStrings.add(randomString);
        }
        ring = new HashRing(randomStrings);
        ProcessChannel<T> channel = new ProcessChannel<>(0, blockingQueueSize, factory.make());
        channels.put(0, channel);
        for (String randomString : randomStrings) {
            int result = ring.map(randomString);

            ProcessChannel<T> channelx = new ProcessChannel<>(result, blockingQueueSize, factory.make());
            channels.put(result, channelx);
        }
    }

    public boolean offer(String uuid, T object) {
        int result = ring.select(uuid);
        ProcessChannel<T> channel = channels.get(result);
        return channel.offer(object);
    }

    public boolean offer(String uuid, T object, long x, TimeUnit unit) throws InterruptedException {
        int result = ring.select(uuid);
        ProcessChannel<T> channel = channels.get(result);
        return channel.offer(object, x, unit);
    }
}
